package com.atguigu.flinkSql2;


import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;


/**
 * @author wky
 * @create 2021-07-22-10:40
 */

//sql 滚动窗口  使用时 必须要放到分组条件里
public class Flink03_Sql_GroupWindow_Hop {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnvironment = StreamTableEnvironment.create(env);
        //将默认时区从格林威治时区改为东八区
        Configuration configuration = tableEnvironment.getConfig().getConfiguration();
        configuration.setString("table.local-time-zone", "GMT");
        //TODO 2.1在创建表时,并指定事件时间
        tableEnvironment.executeSql("create table sensor (" +
                "id string," +
                "ts bigint," +
                "vc int," +
                "t as to_timestamp(from_unixtime(ts/1000,'yyyy-MM-dd HH:mm:ss'))," +
                "watermark for t as t - interval '5' second )" +
                "with(" +
                "'connector' = 'filesystem'," +
                " 'path' = 'src/input/sensor_sql.txt'," +
                "'format' = 'csv'" +
                ")");

        tableEnvironment.executeSql("select id ," +
                " sum(vc) sum_vc " +
                " from sensor " +
                " group by id,tumble(t,interval '3' second)").print();
        //3.查询表,并使用滚动窗口
        tableEnvironment.executeSql(
                "SELECT id, " +
                "  TUMBLE_START(t, INTERVAL '1' minute) as wStart,  " +
                "  TUMBLE_END(t, INTERVAL '1' minute) as wEnd,  " +
                "  SUM(vc) sum_vc " +
                "FROM sensor " +
                "GROUP BY TUMBLE(t, INTERVAL '1' minute), id"
        ).print();

    }
}
